// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/gateio/feed_v4/processor.h"

#include <string>
#include <string_view>

#include <rapidjson/error/en.h>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/gateio/feed_v4/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::gateio::feed_v4::impl {
using namespace std::string_literals;

void GateioFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    LOG(INFO)<<"JSON parse error: "s + GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  // "event":"update" means real data come in, filter subscribe sucesss response message
  if (document.HasMember("channel") &&
      (document["event"].GetString() == "update"s ||
       document["event"].GetString() == "all"s)) {
    std::string channel = document["channel"].GetString();
    if (!CheckChannelFilter(channel)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    if (channel == "spot.trades") {
      const auto& native_symbol = document["result"]["currency_pair"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      Trade* trade = upd_mgr_.GetTradeMutable();
      trade->Clear();
      GateioFeedUpdateParser::ParseTrade(document["result"], trade);
      upd_mgr_.PublishTrade(native_symbol, false);
    } else if (channel == "futures.trades") {
      for (const auto& res : document["result"].GetArray()) {
        const auto& native_symbol = res["contract"].GetString();
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          continue;
        }

        Trade* trade = upd_mgr_.GetTradeMutable();
        trade->Clear();
        GateioFeedUpdateParser::ParseFuturesTrade(res, trade);
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else if (channel == "spot.book_ticker" || channel == "futures.book_ticker") {
      const auto& native_symbol = document["result"]["s"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[std::string(native_symbol)];
        if (!stat.update_received) {
          GateioFeedUpdateParser::ParseBbo(document["result"], book);
          upd_mgr_.PublishBook();
          stat.snapshot_received = true;
        }
      }
    } else if (channel == "spot.order_book_update") {
      const auto& native_symbol = document["result"]["s"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if(GateioFeedUpdateParser::ParseBookUpdate(document["result"], book)){
          upd_mgr_.PublishBook();
        }
      }
    } else if (channel == "futures.order_book_update") {
      const auto& native_symbol = document["result"]["s"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[std::string(native_symbol)];
        if (stat.snapshot_received) {
          GateioFeedUpdateParser::ParseOrderBookUpdate(document["result"], book);
          upd_mgr_.PublishBook();
          stat.update_received = true;
        }
      }
    } else if (channel == "spot.order_book") {
      const auto& native_symbol = document["result"]["s"].GetString();
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          upd_mgr_.onEndPacket(packet);
          return;
        }
        auto* book = upd_mgr_.GetBookMutable(native_symbol);
        if (book != nullptr) {
          if(GateioFeedUpdateParser::ParseBookSnapshot(document["result"], book)){
            upd_mgr_.PublishBook();
          }
        }
    } else if (channel == "futures.order_book") {
      const auto& type = document["event"].GetString();
      if (type == "all"s) {
        const auto& native_symbol = document["result"]["contract"].GetString();
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          upd_mgr_.onEndPacket(packet);
          return;
        }
        auto* book = upd_mgr_.GetBookMutable(native_symbol);
        if (book != nullptr) {
          GateioFeedUpdateParser::ParseOrderBook(document["result"], book);
          upd_mgr_.PublishBook();
        }
      } else if (type == "update"s) {
        auto ts = stoull(document["time"].GetString()) * 1'000'000'000LL;  // ms -> ns
        for (const auto& result : document["result"].GetArray()) {
          const auto& native_symbol = result["c"].GetString();
          if (!upd_mgr_.is_symbol_registered(native_symbol)) {
            upd_mgr_.onEndPacket(packet);
            return;
          }
          auto* book = upd_mgr_.GetBookMutable(native_symbol);
          if (book != nullptr) {
            {
              auto tx = book->NewTransaction(ts);
              double qty = stod(result["s"].GetString());
              if (qty > 0) {
                tx.UpdateBid(stod(result["p"].GetString()), qty);
              } else if (qty < 0) {
                tx.UpdateBid(stod(result["p"].GetString()), - qty);
              }
            }

            upd_mgr_.PublishBook();
          }
        }
      }
    } else {
      LOG(INFO) << "[Gateio] meta msg for " << packet.topic_name << ": " << packet.payload.data();
    }
  } else if (document.HasMember("url")) {
    // rest api feed
    std::string url = document["url"].GetString();
    std::string channel;
    if (StringEndsWith(url, "/contracts")) {
      channel = "index";
    } else if (url.find("/contract_stats?contract=") != std::string::npos) {
      channel = "open_interest";
    } else if (url.find("/candlesticks?") != std::string::npos) {
      channel = "kline_rest";
    }
    if (!CheckChannelFilter(channel)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    if (channel == "index") {
      if (!document["data"].IsArray()) {
        // "data" may is invalid
        return;
      }
      for (const auto& data : document["data"].GetArray()) {
        const auto& native_symbol = data["name"].GetString();
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          upd_mgr_.onEndPacket(packet);
          continue;
        }

        auto* mark_price = upd_mgr_.GetMarkPriceMutable();
        mark_price->timestamp  = packet.timestamp;
        mark_price->mark_price = stod(data["mark_price"].GetString());
        upd_mgr_.PublishMarkPrice(native_symbol);

        auto* index = upd_mgr_.GetIndexMutable();
        index->timestamp = packet.timestamp;
        index->price = stod(data["index_price"].GetString());
        upd_mgr_.PublishIndex(native_symbol);

        auto* funding_rate = upd_mgr_.GetFundingRateMutable();
        funding_rate->timestamp = packet.timestamp;
        funding_rate->funding_rate = stod(data["funding_rate"].GetString());
        upd_mgr_.PublishFundingRate(native_symbol);
      }
    } else if (channel == "open_interest") {
      size_t pos1 = url.find("contract=") + 9;
      size_t pos2 = url.find("&", pos1);
      std::string native_symbol = url.substr(pos1, pos2 - pos1);
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      // only query 1 data for now, but sometimes gateio return 2
      CHECK_LE(document["data"].GetArray().Size(), 2);
      if (document["data"].GetArray().Size() == 0) {
        LOG_FIRST_N(INFO, 100) << "[Gateio] open_interest return empty";
        upd_mgr_.onEndPacket(packet);
        return;
      }
      for (const auto& data : document["data"].GetArray()) {
        // sometimes response has no open_interest field
        if (!data.HasMember("open_interest")) {
          continue;
        }
        auto* open_interest = upd_mgr_.GetOpenInterestMutable();
        open_interest->timestamp = stoll(data["time"].GetString()) * 1'000'000'000LL;
        open_interest->open_interest_qty = stod(data["open_interest"].GetString());
        upd_mgr_.PublishOpenInterest(native_symbol);
      }
    } else if (channel == "kline_rest") {
      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      const std::string& native_symbol = parameters["currency_pair"];
      const std::string& interval_str = parameters["interval"];
      auto interval = GateioFeedUpdateParser::ParseKlineInterval(interval_str);
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (kline != nullptr) {
        const auto& kline_array = document["data"].GetArray();
        if (kline_array.Size() > 10) {
          // first long query
          kline->SnapshotClear();
        } else {
          kline->UpdateClear();
        }

        bool to_publish = false;
        for (const auto& kline_data : kline_array) {
          bool freezed = GateioFeedUpdateParser::Parse1Kline(
              kline_data, kline, packet.timestamp, interval);
          to_publish = to_publish || freezed;
        }

        if (to_publish) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    } else {
      LOG(INFO) << "[Gateio] meta msg for " << packet.topic_name << ": " << packet.payload.data();
    }
  } else {
    LOG(INFO) << "[Gateio] meta msg for " << packet.topic_name << ": " << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}

}  // namespace coin2::exchange::gateio::feed_v4::impl
